package com.central.scheduler.service.process;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.central.common.entity.TaskParams;
import com.central.common.enums.PrivilegeType;
import com.central.common.enums.TaskType;
import com.central.common.feign.*;
import com.central.common.model.*;
import com.central.common.enums.Direct;
import com.central.scheduler.common.Constants;
import com.central.scheduler.common.entity.DateInterval;
import com.central.scheduler.common.entity.TaskNodeRelation;
import com.central.scheduler.common.enums.*;
import com.central.scheduler.common.entity.TaskNode;
import com.central.common.properties.Property;
import com.central.scheduler.common.graph.DAG;
import com.central.scheduler.common.process.ProcessDag;
import com.central.scheduler.common.task.subprocess.SubProcessParameters;
import com.central.scheduler.common.utils.DateUtils;
import com.central.scheduler.common.utils.ParameterUtils;
import com.central.scheduler.dao.entity.TaskNodeParam;
import com.central.scheduler.dao.mapper.*;
import com.central.scheduler.dao.model.*;
import com.central.scheduler.service.quartz.cron.CronUtils;
import com.central.scheduler.common.utils.JSONUtils;
import com.central.scheduler.service.utils.DagHelper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.collect.Maps;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;
import java.util.stream.Collectors;

import static com.central.scheduler.common.Constants.*;
import static java.util.stream.Collectors.toSet;

/**
 * @author Tindy
 * @date 2021/11/1
 * @describe process relative dao that some mappers in this.
 */
@Component
public class ProcessService {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    private final ExecutionStatus[] stateArray = new ExecutionStatus[]{ExecutionStatus.SUBMITTED_SUCCESS,
            ExecutionStatus.RUNNING_EXECUTION,
            ExecutionStatus.DELAY_EXECUTION,
            ExecutionStatus.READY_PAUSE,
            ExecutionStatus.READY_STOP};

    @Autowired
    private ProcessDefinitionMapper processDefineMapper;

    @Autowired
    private ProcessDefinitionLogMapper processDefineLogMapper;

    @Autowired
    private TaskDefinitionMapper taskDefinitionMapper;

    @Autowired
    private ProcessInstanceMapper processInstanceMapper;

    @Autowired
    private ProcessInstanceMapMapper processInstanceMapMapper;

    @Autowired
    private TaskInstanceMapper taskInstanceMapper;

    @Autowired
    private CommandMapper commandMapper;

    @Autowired
    private ScheduleMapper scheduleMapper;

    @Autowired
    private ErrorCommandMapper errorCommandMapper;

    @Autowired
    private ProcessTaskRelationMapper processTaskRelationMapper;
    @Autowired
    private ProcessTaskRelationLogMapper processTaskRelationLogMapper;
    @Autowired
    private EnvironmentMapper environmentMapper;
    @Autowired
    private MasterServerMapper masterServerMapper;
    @Autowired
    private WorkerServerMapper workerServerMapper;
    @Autowired
    private QueueInfoMapper queueInfoMapper;
    @Autowired
    private ProjectService projectService;
    @Autowired
    private UserService userService;
    @Autowired
    private DataManageService dataManageService;
    @Autowired
    private TaskService taskService;
    @Autowired
    private KBMService kbmService;


    /**
     * find task instance by id
     *
     * @param taskId task id
     * @return task intance
     */
    public TaskInstance findTaskInstanceById(Long taskId) {
        return taskInstanceMapper.selectById(taskId);
    }

    /**
     * query need failover process instance
     *
     * @param host host
     * @return process instance list
     */
    public List<ProcessInstance> queryNeedFailoverProcessInstances(String host) {
        return processInstanceMapper.queryByHostAndStatus(host, stateArray);
    }

    /**
     * process need failover process instance
     *
     * @param processInstance processInstance
     */
    @Transactional(rollbackFor = RuntimeException.class)
    public void processNeedFailoverProcessInstances(ProcessInstance processInstance) {
        //1 update processInstance host is null
        processInstance.setHost(Constants.NULL);
        processInstanceMapper.updateById(processInstance);

        ProcessDefinition processDefinition = findProcessDefinition(processInstance.getProcessDefinitionCode(), processInstance.getProcessDefinitionVersion());

        //2 insert into recover command
        Command cmd = new Command();
        cmd.setCommandType(CommandType.RECOVER_TOLERANCE_FAULT_PROCESS);
        cmd.setProcessDefinitionCode(processDefinition.getCode());
        cmd.setProcessDefinitionVersion(processDefinition.getVersion());
        cmd.setProcessInstanceId(processInstance.getId());
        cmd.setCommandParam(String.format("{\"%s\":%d}", Constants.CMD_PARAM_RECOVER_PROCESS_ID_STRING, processInstance.getId()));
        cmd.setUserId(processInstance.getUserId());
        cmd.setUsername(processInstance.getUsername());
        cmd.setProId(processInstance.getProId());
        cmd.setProName(processInstance.getProName());
        createCommand(cmd);
    }

    /**
     * insert one command
     *
     * @param command command
     * @return create result
     */
    public int createCommand(Command command) {
        int result = 0;
        if (command != null) {
            result = commandMapper.insert(command);
        }
        return result;
    }

    /**
     * find process define by code and version.
     *
     * @param processDefinitionCode processDefinitionCode
     * @return process definition
     */
    public ProcessDefinition findProcessDefinition(Long processDefinitionCode, int version) {
        ProcessDefinition processDefinition = processDefineMapper.queryByCode(processDefinitionCode,0l);
        if (processDefinition == null || processDefinition.getVersion() != version) {
            processDefinition = processDefineLogMapper.queryByDefinitionCodeAndVersion(processDefinitionCode, version);
            if (processDefinition != null) {
                processDefinition.setId(0l);
            }
        }
        return processDefinition;
    }

    /**
     * query all need failover task instances by host
     *
     * @param host host
     * @return task instance list
     */
    public List<TaskInstance> queryNeedFailoverTaskInstances(String host) {
        return taskInstanceMapper.queryByHostAndStatus(host,
                stateArray);
    }

    /**
     * find process instance detail by id
     *
     * @param processId processId
     * @return process instance
     */
    public ProcessInstance findProcessInstanceDetailById(Long processId) {
        return processInstanceMapper.selectById(processId);
    }

    /**
     * insert or update task instance
     *
     * @param taskInstance taskInstance
     * @return save task instance result
     */
    public boolean saveTaskInstance(TaskInstance taskInstance) {
        if (taskInstance.getId() != null) {
            return updateTaskInstance(taskInstance);
        } else {
            return createTaskInstance(taskInstance);
        }
    }

    /**
     * insert task instance
     *
     * @param taskInstance taskInstance
     * @return create task instance result
     */
    public boolean createTaskInstance(TaskInstance taskInstance) {
        int count = taskInstanceMapper.insert(taskInstance);
        return count > 0;
    }

    /**
     * update task instance
     *
     * @param taskInstance taskInstance
     * @return update task instance result
     */
    public boolean updateTaskInstance(TaskInstance taskInstance) {
        int count = taskInstanceMapper.updateById(taskInstance);
        return count > 0;
    }

    public Map<ProcessInstance, TaskInstance> notifyProcessList(Long processId, Long taskId) {
        HashMap<ProcessInstance, TaskInstance> processTaskMap = new HashMap<>();
        //find sub tasks
        ProcessInstanceMap processInstanceMap = processInstanceMapMapper.queryBySubProcessId(processId);
        if (processInstanceMap == null) {
            return processTaskMap;
        }
        ProcessInstance fatherProcess = this.findProcessInstanceById(processInstanceMap.getParentProcessInstanceId());
        TaskInstance fatherTask = this.findTaskInstanceById(processInstanceMap.getParentTaskInstanceId());

        if (fatherProcess != null) {
            processTaskMap.put(fatherProcess, fatherTask);
        }
        return processTaskMap;
    }

    /**
     * find process instance by id
     *
     * @param processId processId
     * @return process instance
     */
    public ProcessInstance findProcessInstanceById(Long processId) {
        return processInstanceMapper.selectById(processId);
    }

    /**
     * get command page
     *
     * @param pageSize
     * @param pageNumber
     * @return
     */
    public List<Command> findCommandPage(int pageSize, int pageNumber) {
        return commandMapper.queryCommandPage(pageSize, pageNumber * pageSize);
    }

    /**
     * handle Command (construct ProcessInstance from Command) , wrapped in transaction
     *
     * @param logger                     logger
     * @param host                       host
     * @param command                    found command
     * @param processDefinitionCacheMaps
     * @return process instance
     */
    public ProcessInstance handleCommand(Logger logger, String host, Command command, HashMap<String, ProcessDefinition> processDefinitionCacheMaps) {
        ProcessInstance processInstance = constructProcessInstance(command, host, processDefinitionCacheMaps);
        // cannot construct process instance, return null
        if (processInstance == null) {
            logger.error("scan command, command parameter is error: {}", command);
            moveToErrorCommand(command, "process instance is null");
            return null;
        }
        processInstance.setCommandType(command.getCommandType());
        processInstance.addHistoryCmd(command.getCommandType());
        saveProcessInstance(processInstance);
        this.setSubProcessParam(processInstance);
        this.commandMapper.deleteById(command.getId());
        return processInstance;
    }

    /**
     * set sub work process parameters.
     * handle sub work process instance, update relation table and command parameters
     * set sub work process flag, extends parent work process command parameters
     *
     * @param subProcessInstance subProcessInstance
     */
    public void setSubProcessParam(ProcessInstance subProcessInstance) {
        String cmdParam = subProcessInstance.getCommandParam();
        if (StringUtils.isEmpty(cmdParam)) {
            return;
        }
        Map<String, String> paramMap = JSONUtils.toMap(cmdParam);
        // write sub process id into cmd param.
        if (paramMap.containsKey(CMD_PARAM_SUB_PROCESS)
                && CMD_PARAM_EMPTY_SUB_PROCESS.equals(paramMap.get(CMD_PARAM_SUB_PROCESS))) {
            paramMap.remove(CMD_PARAM_SUB_PROCESS);
            paramMap.put(CMD_PARAM_SUB_PROCESS, String.valueOf(subProcessInstance.getId()));
            subProcessInstance.setCommandParam(JSONUtils.toJsonString(paramMap));
            subProcessInstance.setIsSubProcess(Flag.YES);
            this.saveProcessInstance(subProcessInstance);
        }
        // copy parent instance user def params to sub process..
        String parentInstanceId = paramMap.get(CMD_PARAM_SUB_PROCESS_PARENT_INSTANCE_ID);
        if (StringUtils.isNotEmpty(parentInstanceId)) {
            ProcessInstance parentInstance = findProcessInstanceDetailById(Long.parseLong(parentInstanceId));
            if (parentInstance != null) {
                subProcessInstance.setGlobalParams(
                        joinGlobalParams(parentInstance.getGlobalParams(), subProcessInstance.getGlobalParams()));
                this.saveProcessInstance(subProcessInstance);
            } else {
                logger.error("sub process command params error, cannot find parent instance: {} ", cmdParam);
            }
        }
        ProcessInstanceMap processInstanceMap = JSONUtils.parseObject(cmdParam, ProcessInstanceMap.class);
        if (processInstanceMap == null || processInstanceMap.getParentProcessInstanceId()==null) {
            return;
        }
        // update sub process id to process map table
        processInstanceMap.setProcessInstanceId(subProcessInstance.getId());

        this.updateWorkProcessInstanceMap(processInstanceMap);
    }

    /**
     * update work process instance map
     *
     * @param processInstanceMap processInstanceMap
     * @return update process instance result
     */
    public int updateWorkProcessInstanceMap(ProcessInstanceMap processInstanceMap) {
        return processInstanceMapMapper.update(processInstanceMap,new QueryWrapper<ProcessInstanceMap>()
                .eq("parent_process_instance_id",processInstanceMap.getParentProcessInstanceId())
                .eq("parent_task_instance_id",processInstanceMap.getParentTaskInstanceId()));
    }

    /**
     * join parent global params into sub process.
     * only the keys doesn't in sub process global would be joined.
     *
     * @param parentPropertyList parentPropertyList
     * @param subPropertyList    subPropertyList
     * @return global params join
     */
    private List<Property> joinGlobalParams(List<Property> parentPropertyList, List<Property> subPropertyList) {
        if(parentPropertyList!=null){
            if(subPropertyList!=null){
                Map<String, String> subMap = subPropertyList.stream().collect(Collectors.toMap(Property::getProp, Property::getValue));

                for (Property parent : parentPropertyList) {
                    if (!subMap.containsKey(parent.getProp())) {
                        subPropertyList.add(parent);
                    }
                }
            }else{
                subPropertyList=parentPropertyList;
            }
        }
        return subPropertyList;
    }

    /**
     * insert or update work process instance to data base
     *
     * @param processInstance processInstance
     */
    public void saveProcessInstance(ProcessInstance processInstance) {
        if (processInstance == null) {
            logger.error("save error, process instance is null!");
            return;
        }
        if (processInstance.getId()!=null && processInstance.getId()!= 0) {
            processInstanceMapper.updateById(processInstance);
        } else {
            processInstanceMapper.insert(processInstance);
        }
    }

    /**
     * save error command, and delete original command
     *
     * @param command command
     * @param message message
     */
    public void moveToErrorCommand(Command command, String message) {
        ErrorCommand errorCommand = new ErrorCommand(command, message);
        this.errorCommandMapper.insert(errorCommand);
        this.commandMapper.deleteById(command.getId());
    }

    /**
     * construct process instance according to one command.
     *
     * @param command                    command
     * @param host                       host
     * @param processDefinitionCacheMaps
     * @return process instance
     */
    private ProcessInstance constructProcessInstance(Command command, String host, HashMap<String, ProcessDefinition> processDefinitionCacheMaps) {
        ProcessInstance processInstance;
        ProcessDefinition processDefinition;
        CommandType commandType = command.getCommandType();
        String key = String.format("%d-%d", command.getProcessDefinitionCode(), command.getProcessDefinitionVersion());
        if (processDefinitionCacheMaps.containsKey(key)) {
            processDefinition = processDefinitionCacheMaps.get(key);
        } else {
            processDefinition = this.findProcessDefinition(command.getProcessDefinitionCode(), command.getProcessDefinitionVersion());
            if (processDefinition != null) {
                processDefinitionCacheMaps.put(key, processDefinition);
            }
        }
        if (processDefinition == null) {
            logger.error("cannot find the work process define! define code : {}", command.getProcessDefinitionCode());
            return null;
        }
        Map<String, String> cmdParam = JSONUtils.toMap(command.getCommandParam());
        Long processInstanceId = command.getProcessInstanceId();
        if (processInstanceId == null) {
            processInstance = generateNewProcessInstance(processDefinition, command, cmdParam);
        } else {
            processInstance = this.findProcessInstanceDetailById(processInstanceId);
            if (processInstance == null) {
                return processInstance;
            }
        }
        processInstance.setUserId(command.getUserId());
        processInstance.setUsername(command.getUsername());
        processInstance.setProName(command.getProName());
        if (cmdParam != null) {
            CommandType commandTypeIfComplement = getCommandTypeIfComplement(processInstance, command);
            // reset global params while repeat running is needed by cmdParam
            if (commandTypeIfComplement == CommandType.REPEAT_RUNNING) {
                setGlobalParamIfCommanded(processDefinition, cmdParam);
            }

            // Recalculate global parameters after rerun.
            processInstance.setGlobalParams(ParameterUtils.curingGlobalParams(
                    processDefinition.getGlobalParamMap(),
                    processDefinition.getGlobalParams(),
                    commandTypeIfComplement,
                    processInstance.getScheduleTime()));
            processInstance.setProcessDefinition(processDefinition);
        }
        //reset command parameter
        if (processInstance.getCommandParam() != null) {
            Map<String, String> processCmdParam = JSONUtils.toMap(processInstance.getCommandParam());
            for (Map.Entry<String, String> entry : processCmdParam.entrySet()) {
                if (!cmdParam.containsKey(entry.getKey())) {
                    cmdParam.put(entry.getKey(), entry.getValue());
                }
            }
        }
        // reset command parameter if sub process
        if (cmdParam != null && cmdParam.containsKey(CMD_PARAM_SUB_PROCESS)) {
            processInstance.setCommandParam(command.getCommandParam());
        }
        if (Boolean.FALSE.equals(checkCmdParam(command, cmdParam))) {
            logger.error("command parameter check failed!");
            return null;
        }
        if (command.getScheduleTime() != null) {
            processInstance.setScheduleTime(command.getScheduleTime());
        }
        processInstance.setHost(host);
        ExecutionStatus runStatus = ExecutionStatus.RUNNING_EXECUTION;
        int runTime = processInstance.getRunTimes();
        switch (commandType) {
            case START_PROCESS:
                break;
            case START_FAILURE_TASK_PROCESS:
                // find failed tasks and init these tasks
                List<Long> failedList = this.findTaskIdByInstanceState(processInstance.getId(), ExecutionStatus.FAILURE);
                List<Long> toleranceList = this.findTaskIdByInstanceState(processInstance.getId(), ExecutionStatus.NEED_FAULT_TOLERANCE);
                List<Long> killedList = this.findTaskIdByInstanceState(processInstance.getId(), ExecutionStatus.KILL);
                cmdParam.remove(Constants.CMD_PARAM_RECOVERY_START_NODE_STRING);

                failedList.addAll(killedList);
                failedList.addAll(toleranceList);
                for (Long taskId : failedList) {
                    initTaskInstance(this.findTaskInstanceById(taskId));
                }
                cmdParam.put(Constants.CMD_PARAM_RECOVERY_START_NODE_STRING,
                        String.join(Constants.COMMA, convertLongListToString(failedList)));
                processInstance.setCommandParam(JSONUtils.toJsonString(cmdParam));
                processInstance.setRunTimes(runTime + 1);
                break;
            case START_CURRENT_TASK_PROCESS:
                break;
            case RECOVER_WAITING_THREAD:
                break;
            case RECOVER_SUSPENDED_PROCESS:
                // find pause tasks and init task's state
                cmdParam.remove(Constants.CMD_PARAM_RECOVERY_START_NODE_STRING);
                List<Long> suspendedNodeList = this.findTaskIdByInstanceState(processInstance.getId(), ExecutionStatus.PAUSE);
                List<Long> stopNodeList = findTaskIdByInstanceState(processInstance.getId(),
                        ExecutionStatus.KILL);
                suspendedNodeList.addAll(stopNodeList);
                for (Long taskId : suspendedNodeList) {
                    // initialize the pause state
                    initTaskInstance(this.findTaskInstanceById(taskId));
                }
                cmdParam.put(Constants.CMD_PARAM_RECOVERY_START_NODE_STRING, String.join(",", convertLongListToString(suspendedNodeList)));
                processInstance.setCommandParam(JSONUtils.toJsonString(cmdParam));
                processInstance.setRunTimes(runTime + 1);
                break;
            case RECOVER_TOLERANCE_FAULT_PROCESS:
                // recover tolerance fault process
                processInstance.setRecovery(Flag.YES);
                runStatus = processInstance.getState();
                break;
            case COMPLEMENT_DATA:
                // delete all the valid tasks when complement data if id is not null
                if (processInstance.getId() != null) {
                    List<TaskInstance> taskInstanceList = this.findValidTaskListByProcessId(processInstance.getId());
                    for (TaskInstance taskInstance : taskInstanceList) {
                        taskInstance.setFlag(Flag.NO);
                        this.updateTaskInstance(taskInstance);
                    }
                }
                break;
            case REPEAT_RUNNING:
                // delete the recover task names from command parameter
                if (cmdParam.containsKey(Constants.CMD_PARAM_RECOVERY_START_NODE_STRING)) {
                    cmdParam.remove(Constants.CMD_PARAM_RECOVERY_START_NODE_STRING);
                    processInstance.setCommandParam(JSONUtils.toJsonString(cmdParam));
                }
                // delete all the valid tasks when repeat running
                List<TaskInstance> validTaskList = findValidTaskListByProcessId(processInstance.getId());
                for (TaskInstance taskInstance : validTaskList) {
                    taskInstance.setFlag(Flag.NO);
                    updateTaskInstance(taskInstance);
                }
                processInstance.setStartTime(new Date());
                processInstance.setEndTime(null);
                processInstance.setRunTimes(runTime + 1);
                initComplementDataParam(processDefinition, processInstance, cmdParam);
                break;
            case SCHEDULER:
                break;
            default:
                break;
        }
        processInstance.setState(runStatus);
        return processInstance;
    }

    /**
     * initialize complement data parameters
     *
     * @param processDefinition processDefinition
     * @param processInstance   processInstance
     * @param cmdParam          cmdParam
     */
    private void initComplementDataParam(ProcessDefinition processDefinition,
                                         ProcessInstance processInstance,
                                         Map<String, String> cmdParam) {
        if (!processInstance.isComplementData()) {
            return;
        }

        Date start = DateUtils.stringToDate(cmdParam.get(CMDPARAM_COMPLEMENT_DATA_START_DATE));
        Date end = DateUtils.stringToDate(cmdParam.get(CMDPARAM_COMPLEMENT_DATA_END_DATE));
        List<Schedule> listSchedules = queryReleaseSchedulerListByProcessDefinitionCode(processInstance.getProcessDefinitionCode());
        List<Date> complementDate = CronUtils.getSelfFireDateList(start, end, listSchedules);

        if (complementDate.size() > 0
                && Flag.NO == processInstance.getIsSubProcess()) {
            processInstance.setScheduleTime(complementDate.get(0));
        }
        processInstance.setGlobalParams(ParameterUtils.curingGlobalParams(
                processDefinition.getGlobalParamMap(),
                processDefinition.getGlobalParams(),
                CommandType.COMPLEMENT_DATA, processInstance.getScheduleTime()));
    }

    /**
     * query Schedule by processDefinitionCode
     *
     * @param processDefinitionCode processDefinitionCode
     * @see Schedule
     */
    public List<Schedule> queryReleaseSchedulerListByProcessDefinitionCode(long processDefinitionCode) {
        return scheduleMapper.queryReleaseSchedulerListByProcessDefinitionCode(processDefinitionCode);
    }

    /**
     * convert Long list to string list
     *
     * @param longList longList
     * @return string list
     */
    public List<String> convertLongListToString(List<Long> longList) {
        if (longList == null) {
            return new ArrayList<>();
        }
        List<String> result = new ArrayList<>(longList.size());
        for (Long longVar : longList) {
            result.add(String.valueOf(longVar));
        }
        return result;
    }

    /**
     * initialize task instance
     *
     * @param taskInstance taskInstance
     */
    private void initTaskInstance(TaskInstance taskInstance) {

        if (!taskInstance.isSubProcess()
                && (taskInstance.getState().typeIsCancel() || taskInstance.getState().typeIsFailure())) {
            taskInstance.setFlag(Flag.NO);
            updateTaskInstance(taskInstance);
            return;
        }
        taskInstance.setState(ExecutionStatus.SUBMITTED_SUCCESS);
        updateTaskInstance(taskInstance);
    }

    /**
     * get id list by task state
     *
     * @param instanceId instanceId
     * @param state      state
     * @return task instance states
     */
    public List<Long> findTaskIdByInstanceState(Long instanceId, ExecutionStatus state) {
        return taskInstanceMapper.queryTaskByProcessIdAndState(instanceId, state.ordinal());
    }

    /**
     * find valid task list by process definition id
     *
     * @param processInstanceId processInstanceId
     * @return task instance list
     */
    public List<TaskInstance> findValidTaskListByProcessId(Long processInstanceId) {
        return taskInstanceMapper.findValidTaskListByProcessId(processInstanceId, Flag.YES);
    }

    /**
     * find previous task list by work process id
     *
     * @param processInstanceId processInstanceId
     * @return task instance list
     */
    public List<TaskInstance> findPreviousTaskListByWorkProcessId(Long processInstanceId) {
        return taskInstanceMapper.findValidTaskListByProcessId(processInstanceId, Flag.NO);
    }

    /**
     * check command parameters is valid
     *
     * @param command  command
     * @param cmdParam cmdParam map
     * @return whether command param is valid
     */
    private Boolean checkCmdParam(Command command, Map<String, String> cmdParam) {
        if (command.getTaskDependType() == TaskDependType.TASK_ONLY || command.getTaskDependType() == TaskDependType.TASK_PRE) {
            if (cmdParam == null
                    || !cmdParam.containsKey(Constants.CMD_PARAM_START_NODE_NAMES)
                    || cmdParam.get(Constants.CMD_PARAM_START_NODE_NAMES).isEmpty()) {
                logger.error("command node depend type is {}, but start nodes is null ", command.getTaskDependType());
                return false;
            }
        }
        return true;
    }

    /**
     * generate a new work process instance from command.
     *
     * @param processDefinition processDefinition
     * @param command           command
     * @param cmdParam          cmdParam map
     * @return process instance
     */
    private ProcessInstance generateNewProcessInstance(ProcessDefinition processDefinition,
                                                       Command command,
                                                       Map<String, String> cmdParam) {
        ProcessInstance processInstance = new ProcessInstance(processDefinition);
        processInstance.setProcessDefinitionCode(processDefinition.getCode());
        processInstance.setProcessDefinitionVersion(processDefinition.getVersion());
        processInstance.setState(ExecutionStatus.RUNNING_EXECUTION);
        processInstance.setRecovery(Flag.NO);
        processInstance.setStartTime(new Date());
        processInstance.setRunTimes(1);
        processInstance.setMaxTryTimes(0);
        processInstance.setProcessDefinition(processDefinition);
        processInstance.setCommandParam(command.getCommandParam());
        processInstance.setCommandType(command.getCommandType());
        processInstance.setIsSubProcess(Flag.NO);
        processInstance.setTaskDependType(command.getTaskDependType());
        processInstance.setFailureStrategy(command.getFailureStrategy());
        processInstance.setUserId(command.getUserId());
        processInstance.setProId(command.getProId());
        WarningType warningType = command.getWarningType() == null ? WarningType.NONE : command.getWarningType();
        processInstance.setWarningType(warningType);
        Long warningGroupId = command.getWarningGroupId() == null ? 0 : command.getWarningGroupId();
        processInstance.setWarningGroupId(warningGroupId);
        processInstance.setDryRun(command.getDryRun());

        if (command.getScheduleTime() != null) {
            processInstance.setScheduleTime(command.getScheduleTime());
        }
        processInstance.setCommandStartTime(command.getStartTime());

        // reset global params while there are start parameters
        setGlobalParamIfCommanded(processDefinition, cmdParam);

        // curing global params
        processInstance.setGlobalParams(ParameterUtils.curingGlobalParams(
                processDefinition.getGlobalParamMap(),
                processDefinition.getGlobalParams(),
                getCommandTypeIfComplement(processInstance, command),
                processInstance.getScheduleTime()));

        // set process instance priority
        processInstance.setProcessInstancePriority(command.getProcessInstancePriority());
        processInstance.setTimeout(processDefinition.getTimeout());
        return processInstance;
    }

    /**
     * return complement data if the process start with complement data
     *
     * @param processInstance processInstance
     * @param command         command
     * @return command type
     */
    private CommandType getCommandTypeIfComplement(ProcessInstance processInstance, Command command) {
        if (CommandType.COMPLEMENT_DATA == processInstance.getCmdTypeIfComplement()) {
            return CommandType.COMPLEMENT_DATA;
        } else {
            return command.getCommandType();
        }
    }

    private void setGlobalParamIfCommanded(ProcessDefinition processDefinition, Map<String, String> cmdParam) {
        // get start params from command param
        Map<String, String> startParamMap = new HashMap<>();
        if (cmdParam != null && cmdParam.containsKey(Constants.CMD_PARAM_START_PARAMS)) {
            String startParamJson = cmdParam.get(Constants.CMD_PARAM_START_PARAMS);
            startParamMap = JSONUtils.toMap(startParamJson);
        }
        Map<String, String> fatherParamMap = new HashMap<>();
        if (cmdParam != null && cmdParam.containsKey(Constants.CMD_PARAM_FATHER_PARAMS)) {
            String fatherParamJson = cmdParam.get(Constants.CMD_PARAM_FATHER_PARAMS);
            fatherParamMap = JSONUtils.toMap(fatherParamJson);
        }
        startParamMap.putAll(fatherParamMap);
        // set start param into global params
        if (startParamMap.size() > 0
                && processDefinition.getGlobalParamMap() != null) {
            for (Map.Entry<String, String> param : processDefinition.getGlobalParamMap().entrySet()) {
                String val = startParamMap.get(param.getKey());
                if (val != null) {
                    param.setValue(val);
                }
            }
        }
    }

    /**
     * find process task relation list by proId and processDefinitionCode
     */
    public List<ProcessTaskRelation> findRelationByCode(long processDefinitionCode) {
        return processTaskRelationMapper.selectList(new QueryWrapper<ProcessTaskRelation>().eq("process_definition_code",processDefinitionCode));

    }
    /**
     * Use temporarily before refactoring taskNode
     */
    public List<TaskNode> transformTask(List<TaskNodeParam> taskNodeParams,List<ProcessTaskRelation> taskRelationList, Map<String, TaskDefinition> taskDefinitionMap) {
        Map<String, List<String>> taskCodeMap = new HashMap<>();
        for (ProcessTaskRelation processTaskRelation : taskRelationList) {
            taskCodeMap.compute(processTaskRelation.getPostTaskName(), (k, v) -> {
                if (v == null) {
                    v = new ArrayList<>();
                }
                if (processTaskRelation.getPreTaskName() != null) {
                    v.add(processTaskRelation.getPreTaskName());
                }
                return v;
            });
        }
        if (taskDefinitionMap.isEmpty()) {
            taskDefinitionMap = genTaskDefinitionMap(taskNodeParams);
        }
        List<TaskNode> taskNodeList = new ArrayList<>();
        for (TaskNodeParam taskNodeParam : taskNodeParams) {
            TaskNode taskNode = new TaskNode();
            TaskDefinition taskDefinition = taskDefinitionMap.get(taskNodeParam.getName());
            if(taskDefinition!=null){
                taskNode.setTaskDefinitionId(taskDefinition.getId());
                taskNode.setTaskDefinedLogId(taskDefinition.getTaskDefinedLogId());
                taskNode.setVersion(taskDefinition.getVersion());
                taskNode.setDesc(taskDefinition.getDescription());
                taskNode.setMaxRetryTimes(taskDefinition.getTaskParams().getFailRetryTimes());
                taskNode.setRetryInterval(taskDefinition.getTaskParams().getFailRetryInterval());
                taskNode.setResourceIds(taskDefinition.getTaskParams().getResourceIds());
                taskNode.setTaskContent(taskDefinition.getTaskContent());
                taskNode.setLocalParams(taskDefinition.getTaskParams().getLocalParams());
                taskNode.setWorkerGroupId(taskDefinition.getTaskParams().getWorkerGroupId());
                taskNode.setEnvironmentId(taskDefinition.getTaskParams().getEnvironmentId());
            }else {
                taskNode.setTaskDefinitionId(0l);
                taskNode.setWorkerGroupId(0l);
            }
            taskNode.setName(taskNodeParam.getName());
            taskNode.setType(taskNodeParam.getTaskType());
            taskNode.setRunFlag(taskNodeParam.getFlag() == Flag.YES ? Constants.FLOWNODE_RUN_FLAG_NORMAL : Constants.FLOWNODE_RUN_FLAG_FORBIDDEN);
            switch (taskNodeParam.getTaskType()){
                case CONDITIONS:
                    taskNode.setRelationParameters(taskNodeParam.getConditionsParameters());
                    break;
                case SWITCH:
                    taskNode.setRelationParameters(taskNodeParam.getSwitchParameters());
                    break;
                case DEPENDENT:
                    taskNode.setRelationParameters(taskNodeParam.getDependentParameters());
                    break;
                case SUB_PROCESS:
                    taskNode.setRelationParameters(taskNodeParam.getSubProcessParameters());
                    break;
                default:
                    break;
            }
            taskNode.setTaskInstancePriority(taskNodeParam.getTaskPriority());
            taskNode.setTimeout(taskNodeParam.getTaskTimeoutParameter());
            taskNode.setDelayTime(taskNodeParam.getDelayTime());
            taskNode.setDepList(taskCodeMap.get(taskNodeParam.getName()));
            taskNodeList.add(taskNode);
        }
        return taskNodeList;
    }

    private HashMap<String, TaskNodeParam> genTaskNodeParamMap(List<TaskNodeParam> taskNodeParams) {
        HashMap<String, TaskNodeParam> taskNodeParamMap = new HashMap<>();
        for (TaskNodeParam taskNodeParam : taskNodeParams) {
            taskNodeParamMap.put(taskNodeParam.getName(),taskNodeParam);
        }
        return taskNodeParamMap;
    }


    /**
     * get an environment
     * use the id of the environment to find a environment.
     *
     * @param environmentId environmentId
     * @return Environment
     */
    public Environment findEnvironmentById(Long environmentId) {
        Environment environment = null;
        if (environmentId >= 0) {
            environment = environmentMapper.selectById(environmentId);
        }
        return environment;
    }

    /**
     * update process instance
     *
     * @param processInstance processInstance
     * @return update process instance result
     */
    public int updateProcessInstance(ProcessInstance processInstance) {
        return processInstanceMapper.updateById(processInstance);
    }

    public Map<String,TaskDefinition> genTaskDefinitionMap(List<TaskNodeParam> taskNodeParams) {
        HashMap<String, TaskDefinition> taskDefinitionMap = new HashMap<>();

        for (TaskNodeParam taskNodeParam : taskNodeParams) {
            if(taskNodeParam.getTaskDefinitionId()!=null){
                TaskDefinition taskDefinition = taskDefinitionMapper.selectById(taskNodeParam.getTaskDefinitionId());
                taskDefinitionMap.put(taskNodeParam.getName(),taskDefinition);
            }
        }
        return taskDefinitionMap;
    }


    /**
     * create recovery waiting thread command when thread pool is not enough for the process instance.
     * sub work process instance need not to create recovery command.
     * create recovery waiting thread  command and delete origin command at the same time.
     * if the recovery command is exists, only update the field update_time
     *
     * @param originCommand   originCommand
     * @param processInstance processInstance
     */
    public void createRecoveryWaitingThreadCommand(Command originCommand, ProcessInstance processInstance) {

        // sub process doesnot need to create wait command
        if (processInstance.getIsSubProcess() == Flag.YES) {
            if (originCommand != null) {
                commandMapper.deleteById(originCommand.getId());
            }
            return;
        }
        Map<String, String> cmdParam = new HashMap<>();
        cmdParam.put(Constants.CMD_PARAM_RECOVERY_WAITING_THREAD, String.valueOf(processInstance.getId()));
        // process instance quit by "waiting thread" state
        if (originCommand == null) {
            Command command = new Command(
                    CommandType.RECOVER_WAITING_THREAD,
                    processInstance.getTaskDependType(),
                    processInstance.getFailureStrategy(),
                    processInstance.getUserId(),
                    processInstance.getUsername(),
                    processInstance.getProId(),
                    processInstance.getProName(),
                    processInstance.getProcessDefinition().getCode(),
                    JSONUtils.toJsonString(cmdParam),
                    processInstance.getWarningType(),
                    processInstance.getWarningGroupId(),
                    processInstance.getScheduleTime(),
                    processInstance.getProcessInstancePriority(),
                    processInstance.getDryRun(),
                    processInstance.getId(),
                    processInstance.getProcessDefinitionVersion()
            );
            saveCommand(command);
            return;
        }

        // update the command time if current command if recover from waiting
        if (originCommand.getCommandType() == CommandType.RECOVER_WAITING_THREAD) {
            originCommand.setUpdateTime(new Date());
            saveCommand(originCommand);
        } else {
            // delete old command and create new waiting thread command
            commandMapper.deleteById(originCommand.getId());
            originCommand.setId(0l);
            originCommand.setCommandType(CommandType.RECOVER_WAITING_THREAD);
            originCommand.setUpdateTime(new Date());
            originCommand.setCommandParam(JSONUtils.toJsonString(cmdParam));
            originCommand.setProcessInstancePriority(processInstance.getProcessInstancePriority());
            saveCommand(originCommand);
        }
    }

    /**
     * insert or update command
     *
     * @param command command
     * @return save command result
     */
    public int saveCommand(Command command) {
        if (command.getId() != null) {
            return commandMapper.updateById(command);
        } else {
            return commandMapper.insert(command);
        }
    }


    public void updateNodeInfoOnDB(String host, NodeType nodeType, String path) {
        switch (nodeType) {
            case MASTER:
                MasterServer masterServer = masterServerMapper.findByHost(host);
                if (masterServer != null) {
                    masterServerMapper.updateById(masterServer);
                } else {
                    masterServer = new MasterServer();
                    masterServer.setName(host);
                    masterServer.setHost(host);
                    masterServer.setStatus(1);
                    masterServerMapper.insert(masterServer);
                }
                break;
            case WORKER:
                WorkerServer workerServer = workerServerMapper.findByHost(host);
                if (workerServer != null) {
                    workerServerMapper.updateById(workerServer);
                } else {
                    workerServer = new WorkerServer();
                    workerServer.setName(host);
                    workerServer.setHost(host);
                    workerServer.setStatus(1);
                    String workerType = getWorkerTypeByEventDataPath(path);
                    workerServerMapper.insert(workerServer);
                }
                break;
            default:
                break;
        }
    }

    private String getWorkerTypeByEventDataPath(String path) {
        String[] pathArray = path.split(SINGLE_SLASH);
        if (pathArray.length < 2) {
            logger.error("parse ip error: {}", path);
            return "";
        }
        return pathArray[pathArray.length - 2];
    }

    /**
     * save processDefinition (including create or update processDefinition)
     */
    public int saveProcessDefine(ProcessDefinition processDefinition, Boolean isFromProcessDefine) {
        ProcessDefinitionLog processDefinitionLog = new ProcessDefinitionLog(processDefinition);
        Integer version = processDefineLogMapper.queryMaxVersionForDefinition(processDefinition.getCode());
        int insertVersion = version == null || version == 0 ? Constants.VERSION_FIRST : version + 1;
        processDefinitionLog.setVersion(insertVersion);
        processDefinitionLog.setReleaseState(ReleaseState.OFFLINE);
        processDefinition.setVersion(insertVersion);
        processDefinition.setReleaseState(isFromProcessDefine ? ReleaseState.OFFLINE : ReleaseState.ONLINE);
        int insertLog = processDefineLogMapper.insert(processDefinitionLog);
        int result;
        if (processDefinition.getId()==null || 0 == processDefinition.getId()) {
            result = processDefineMapper.insert(processDefinition);
        } else {
            processDefinitionLog.setId(processDefinition.getId());
            result = processDefineMapper.updateById(processDefinition);
        }
        return (insertLog & result) > 0 ? insertVersion : 0;
    }

    /**
     * save task relations
     */
    public int saveTaskRelation(long proId, long processDefinitionCode, int processDefinitionVersion,
                                List<ProcessTaskRelationLog> taskRelationList) {
        Date now = new Date();
        for (ProcessTaskRelationLog processTaskRelationLog : taskRelationList) {
            processTaskRelationLog.setProId(proId);
            processTaskRelationLog.setProcessDefinitionCode(processDefinitionCode);
            processTaskRelationLog.setProcessDefinitionVersion(processDefinitionVersion);
        }
        List<ProcessTaskRelation> processTaskRelationList = processTaskRelationMapper.selectList(new QueryWrapper<ProcessTaskRelation>().eq("process_definition_code",processDefinitionCode));
        if (!processTaskRelationList.isEmpty()) {
            Set<Integer> processTaskRelationSet = processTaskRelationList.stream().map(ProcessTaskRelation::hashCode).collect(toSet());
            Set<Integer> taskRelationSet = taskRelationList.stream().map(ProcessTaskRelationLog::hashCode).collect(toSet());
            if (CollectionUtils.isEqualCollection(processTaskRelationSet, taskRelationSet)) {
                return Constants.EXIT_CODE_SUCCESS;
            }
            processTaskRelationMapper.delete(new QueryWrapper<ProcessTaskRelation>().eq("process_definition_code",processDefinitionCode));
        }

        int result = processTaskRelationMapper.batchInsert(taskRelationList);
        int resultLog = processTaskRelationLogMapper.batchInsert(taskRelationList);
        return (result & resultLog) > 0 ? Constants.EXIT_CODE_SUCCESS : Constants.EXIT_CODE_FAILURE;
    }

    public TaskDefinition findTaskDefinition(long taskDefinitionId) {
        return taskDefinitionMapper.selectById(taskDefinitionId);
    }

    public SysProject queryProjectByProcessInstance(ProcessInstance processInstance) {
        ArrayList<Long> list = new ArrayList<>();
        list.add(processInstance.getProId());
        List<SysProject> pros = projectService.findProByProIds(list);
        if(!pros.isEmpty()){
            return pros.get(0);
        }else {
            return null;
        }
    }

    public String queryUserByProcessInstance(ProcessInstance processInstance) {
        Long userId = processInstance.getUserId();
        String username = userService.selectById(userId).getUsername();
        return username;
    }
    /**
     * package task instance，associate processInstance and processDefine
     *
     * @param taskInstance taskInstance
     * @return task instance
     */
    public void setTaskInstanceDetail(TaskInstance taskInstance) {
        // get process instance
        ProcessInstance processInstance = findProcessInstanceDetailById(taskInstance.getProcessInstanceId());
        // get process define
        ProcessDefinition processDefine = findProcessDefinition(processInstance.getProcessDefinitionCode(),
                processInstance.getProcessDefinitionVersion());
        taskInstance.setProcessInstance(processInstance);
        taskInstance.setProcessDefine(processDefine);
        TaskDefinition taskDefinition = taskDefinitionMapper.selectById(taskInstance.getTaskDefinitionId());
        if(taskDefinition!=null){
            updateTaskDefinitionResources(taskDefinition);
            taskInstance.setTaskDefine(taskDefinition);
        }
    }
    /**
     * Update {@link ResourceInfo} information in {@link TaskDefinition}
     *
     * @param taskDefinition the given {@link TaskDefinition}
     */
    private void updateTaskDefinitionResources(TaskDefinition taskDefinition) {
        TaskParams taskParams = taskDefinition.getTaskParams();
        if (taskParams != null) {
//            // if contains mainJar field, query resource from database
//            // Flink, Spark, MR
//            if (taskParameters.containsKey("mainJar")) {
//                Object mainJarObj = taskParameters.get("mainJar");
//                ResourceInfo mainJar = JSONUtils.parseObject(
//                        JSONUtils.toJsonString(mainJarObj),
//                        ResourceInfo.class);
//                ResourceInfo resourceInfo = updateResourceInfo(mainJar);
//                if (resourceInfo != null) {
//                    taskParameters.put("mainJar", resourceInfo);
//                }
//            }
//            // update resourceList information
//            if (taskParameters.containsKey("resourceList")) {
//                String resourceListStr = JSONUtils.toJsonString(taskParameters.get("resourceList"));
//                List<ResourceInfo> resourceInfos = JSONUtils.toList(resourceListStr, ResourceInfo.class);
//                List<ResourceInfo> updatedResourceInfos = resourceInfos
//                        .stream()
//                        .map(this::updateResourceInfo)
//                        .filter(Objects::nonNull)
//                        .collect(Collectors.toList());
//                taskParameters.put("resourceList", updatedResourceInfos);
//            }
//            // set task parameters
//            taskDefinition.setTaskParams(JSONUtils.toJsonString(taskParameters));
        }
    }

    /**
     * change task state
     *
     * @param state state
     * @param startTime startTime
     * @param host host
     * @param executePath executePath
     * @param logPath logPath
     */
    public void changeTaskState(TaskInstance taskInstance, ExecutionStatus state, Date startTime, String host,
                                String executePath,
                                String logPath) {
        taskInstance.setState(state);
        taskInstance.setStartTime(startTime);
        taskInstance.setHost(host);
        taskInstance.setExecutePath(executePath);
        taskInstance.setLogPath(logPath);
        saveTaskInstance(taskInstance);
    }

    /**
     * change task state
     *
     * @param state state
     * @param endTime endTime
     * @param varPool varPool
     */
    public void changeTaskState(TaskInstance taskInstance, ExecutionStatus state,
                                Date endTime,
                                int processId,
                                String appIds,
                                List<Property> varPool) {
        taskInstance.setPid(processId);
        taskInstance.setAppLink(appIds);
        taskInstance.setState(state);
        taskInstance.setEndTime(endTime);
        taskInstance.setVarPool(varPool);
        changeOutParam(taskInstance);
        saveTaskInstance(taskInstance);
    }
    /**
     * for show in page of taskInstance
     *
     * @param taskInstance
     */
    public void changeOutParam(TaskInstance taskInstance) {
        if (CollectionUtils.isEmpty(taskInstance.getVarPool())) {
            return;
        }
        List<Property> properties = taskInstance.getVarPool();
        if (CollectionUtils.isEmpty(properties)) {
            return;
        }
        //if the result more than one line,just get the first .
        List<Property> allParam = taskInstance.getLocalParams();
        Map<String, String> outProperty = new HashMap<>();
        for (Property info : properties) {
            if (info.getDirect() == Direct.OUT) {
                outProperty.put(info.getProp(), info.getValue());
            }
        }
        for (Property info : allParam) {
            if (info.getDirect() == Direct.OUT) {
                String paramName = info.getProp();
                info.setValue(outProperty.get(paramName));
            }
        }
    }
    /**
     * retry submit task to db
     *
     * @param taskInstance
     * @param commitRetryTimes
     * @param commitInterval
     * @return
     */
    public TaskInstance submitTask(TaskInstance taskInstance, int commitRetryTimes, int commitInterval) {

        int retryTimes = 1;
        boolean submitDB = false;
        TaskInstance task = null;
        while (retryTimes <= commitRetryTimes) {
            try {
                if (!submitDB) {
                    // submit task to db
                    task = submitTask(taskInstance);
                    if (task != null && task.getId() != null) {
                        submitDB = true;
                        break;
                    }
                }
                if (!submitDB) {
                    logger.error("task commit to db failed , taskId {} has already retry {} times, please check the database", taskInstance.getId(), retryTimes);
                }
                Thread.sleep(commitInterval);
            } catch (Exception e) {
                logger.error("task commit to mysql failed", e);
            }
            retryTimes += 1;
        }
        return task;
    }
    /**
     * submit task to db
     * submit sub process to command
     *
     * @param taskInstance taskInstance
     * @return task instance
     */
    @Transactional(rollbackFor = Exception.class)
    public TaskInstance submitTask(TaskInstance taskInstance) {
        ProcessInstance processInstance = this.findProcessInstanceDetailById(taskInstance.getProcessInstanceId());
        logger.info("start submit task : {}, instance id:{}, state: {}",
                taskInstance.getName(), taskInstance.getProcessInstanceId(), processInstance.getState());
        //submit to db
        TaskInstance task = submitTaskInstanceToDB(taskInstance, processInstance);
        if (task == null) {
            logger.error("end submit task to db error, task name:{}, process id:{} state: {} ",
                    taskInstance.getName(), taskInstance.getProcessInstance(), processInstance.getState());
            return task;
        }
        if (!task.getState().typeIsFinished()) {
            createSubWorkProcess(processInstance, task);
        }

        logger.info("end submit task to db successfully:{} {} state:{} complete, instance id:{} state: {}  ",
                taskInstance.getId(), taskInstance.getName(), task.getState(), processInstance.getId(), processInstance.getState());
        return task;
    }
    /**
     * create sub work process command
     *
     * @param parentProcessInstance parentProcessInstance
     * @param task task
     */
    public void createSubWorkProcess(ProcessInstance parentProcessInstance, TaskInstance task) {
        if (!task.isSubProcess()) {
            return;
        }
        //check create sub work flow firstly
        ProcessInstanceMap instanceMap = findWorkProcessMapByParent(parentProcessInstance.getId(), task.getId());
        if (null != instanceMap && CommandType.RECOVER_TOLERANCE_FAULT_PROCESS == parentProcessInstance.getCommandType()) {
            // recover failover tolerance would not create a new command when the sub command already have been created
            return;
        }
        instanceMap = setProcessInstanceMap(parentProcessInstance, task);
        ProcessInstance childInstance = null;
        if (instanceMap.getProcessInstanceId() != null) {
            childInstance = findProcessInstanceById(instanceMap.getProcessInstanceId());
        }
        Command subProcessCommand = createSubProcessCommand(parentProcessInstance, childInstance, instanceMap, task);
        updateSubProcessDefinitionByParent(parentProcessInstance, subProcessCommand.getProcessDefinitionCode());
        initSubInstanceState(childInstance);
        createCommand(subProcessCommand);
        logger.info("sub process command created: {} ", subProcessCommand);
    }

    /**
     * initialize sub work flow state
     * child instance state would be initialized when 'recovery from pause/stop/failure'
     */
    private void initSubInstanceState(ProcessInstance childInstance) {
        if (childInstance != null) {
            childInstance.setState(ExecutionStatus.RUNNING_EXECUTION);
            updateProcessInstance(childInstance);
        }
    }

    /**
     * update sub process definition
     *
     * @param parentProcessInstance parentProcessInstance
     * @param childDefinitionCode childDefinitionId
     */
    private void updateSubProcessDefinitionByParent(ProcessInstance parentProcessInstance, long childDefinitionCode) {
        ProcessDefinition fatherDefinition = this.findProcessDefinition(parentProcessInstance.getProcessDefinitionCode(),
                parentProcessInstance.getProcessDefinitionVersion());
        ProcessDefinition childDefinition = processDefineMapper.queryByCode(childDefinitionCode,0l);
        if (childDefinition != null && fatherDefinition != null) {
            childDefinition.setWarningGroupId(fatherDefinition.getWarningGroupId());
            processDefineMapper.updateById(childDefinition);
        }
    }
    /**
     * create sub work process command
     */
    public Command createSubProcessCommand(ProcessInstance parentProcessInstance,
                                           ProcessInstance childInstance,
                                           ProcessInstanceMap instanceMap,
                                           TaskInstance task) {
        CommandType commandType = getSubCommandType(parentProcessInstance, childInstance);
        SubProcessParameters relationParameters = (SubProcessParameters) task.getRelationParameters();
        Long childDefineCode= relationParameters.getProcessDefinitionCode();
        ProcessDefinition subProcessDefinition = processDefineMapper.queryByCode(childDefineCode,0l);

        List<Property> allParam = task.getLocalParams();
        Map<String, String> globalMap = this.getGlobalParamMap(parentProcessInstance.getGlobalParams());
        Map<String, String> fatherParams = new HashMap<>();
        if (CollectionUtils.isNotEmpty(allParam)) {
            for (Property info : allParam) {
                fatherParams.put(info.getProp(), globalMap.get(info.getProp()));
            }
        }
        String processParam = getSubWorkFlowParam(instanceMap, parentProcessInstance, fatherParams);
        Long subProcessInstanceId = childInstance == null ? null : childInstance.getId();
        return new Command(
                commandType,
                TaskDependType.TASK_POST,
                parentProcessInstance.getFailureStrategy(),
                parentProcessInstance.getUserId(),
                parentProcessInstance.getUsername(),
                parentProcessInstance.getProId(),
                parentProcessInstance.getProName(),
                subProcessDefinition.getCode(),
                processParam,
                parentProcessInstance.getWarningType(),
                parentProcessInstance.getWarningGroupId(),
                parentProcessInstance.getScheduleTime(),
                parentProcessInstance.getProcessInstancePriority(),
                parentProcessInstance.getDryRun(),
                subProcessInstanceId,
                subProcessDefinition.getVersion()
        );
    }


    /**
     * complement data needs transform parent parameter to child.
     */
    private String getSubWorkFlowParam(ProcessInstanceMap instanceMap, ProcessInstance parentProcessInstance, Map<String, String> fatherParams) {
        // set sub work process command
        String processMapStr = JSONUtils.toJsonString(instanceMap);
        Map<String, String> cmdParam = JSONUtils.toMap(processMapStr);
        if (parentProcessInstance.isComplementData()) {
            Map<String, String> parentParam = JSONUtils.toMap(parentProcessInstance.getCommandParam());
            String endTime = parentParam.get(CMDPARAM_COMPLEMENT_DATA_END_DATE);
            String startTime = parentParam.get(CMDPARAM_COMPLEMENT_DATA_START_DATE);
            cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, endTime);
            cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, startTime);
            processMapStr = JSONUtils.toJsonString(cmdParam);
        }
        if (fatherParams.size() != 0) {
            cmdParam.put(CMD_PARAM_FATHER_PARAMS, JSONUtils.toJsonString(fatherParams));
            processMapStr = JSONUtils.toJsonString(cmdParam);
        }
        return processMapStr;
    }

    public Map<String, String> getGlobalParamMap(List<Property> globalParams) {
        List<Property> propList;
        Map<String, String> globalParamMap = new HashMap<>();
        if (CollectionUtils.isNotEmpty(globalParams)) {
            globalParamMap = globalParams.stream().collect(Collectors.toMap(Property::getProp, Property::getValue));
        }

        return globalParamMap;
    }

    /**
     * get sub work flow command type
     * child instance exist: child command = fatherCommand
     * child instance not exists: child command = fatherCommand[0]
     */
    private CommandType getSubCommandType(ProcessInstance parentProcessInstance, ProcessInstance childInstance) {
        CommandType commandType = parentProcessInstance.getCommandType();
        if (childInstance == null) {
            String fatherHistoryCommand = parentProcessInstance.getHistoryCmd();
            commandType = CommandType.valueOf(fatherHistoryCommand.split(Constants.COMMA)[0]);
        }
        return commandType;
    }

    /**
     * set work process instance map
     * consider o
     * repeat running  does not generate new sub process instance
     * set map {parent instance id, task instance id, 0(child instance id)}
     *
     * @param parentInstance parentInstance
     * @param parentTask parentTask
     * @return process instance map
     */
    private ProcessInstanceMap setProcessInstanceMap(ProcessInstance parentInstance, TaskInstance parentTask) {
        ProcessInstanceMap processMap = findWorkProcessMapByParent(parentInstance.getId(), parentTask.getId());
        if (processMap != null) {
            return processMap;
        }
        if (parentInstance.getCommandType() == CommandType.REPEAT_RUNNING) {
            // update current task id to map
            processMap = findPreviousTaskProcessMap(parentInstance, parentTask);
            if (processMap != null) {
                processMap.setParentTaskInstanceId(parentTask.getId());
                updateWorkProcessInstanceMap(processMap);
                return processMap;
            }
        }
        // new task
        processMap = new ProcessInstanceMap();
        processMap.setParentProcessInstanceId(parentInstance.getId());
        processMap.setParentTaskInstanceId(parentTask.getId());
        createWorkProcessInstanceMap(processMap);
        return processMap;
    }
    /**
     * create work process instance map
     *
     * @param processInstanceMap processInstanceMap
     * @return create process instance result
     */
    public int createWorkProcessInstanceMap(ProcessInstanceMap processInstanceMap) {
        int count = 0;
        if (processInstanceMap != null) {
            return processInstanceMapMapper.insert(processInstanceMap);
        }
        return count;
    }

    /**
     * find previous task work process map.
     *
     * @param parentProcessInstance parentProcessInstance
     * @param parentTask parentTask
     * @return process instance map
     */
    private ProcessInstanceMap findPreviousTaskProcessMap(ProcessInstance parentProcessInstance,
                                                          TaskInstance parentTask) {

        Long preTaskId = 0l;
        List<TaskInstance> preTaskList = this.findPreviousTaskListByWorkProcessId(parentProcessInstance.getId());
        for (TaskInstance task : preTaskList) {
            if (task.getName().equals(parentTask.getName())) {
                preTaskId = task.getId();
                ProcessInstanceMap map = findWorkProcessMapByParent(parentProcessInstance.getId(), preTaskId);
                if (map != null) {
                    return map;
                }
            }
        }
        logger.info("sub process instance is not found,parent task:{},parent instance:{}",
                parentTask.getId(), parentProcessInstance.getId());
        return null;
    }

    /**
     * find work process map by parent process id and parent task id.
     *
     * @param parentWorkProcessId parentWorkProcessId
     * @param parentTaskId parentTaskId
     * @return process instance map
     */
    public ProcessInstanceMap findWorkProcessMapByParent(Long parentWorkProcessId, Long parentTaskId) {
        return processInstanceMapMapper.queryByParentId(parentWorkProcessId, parentTaskId);
    }

    /**
     * submit task to mysql
     *
     * @param taskInstance taskInstance
     * @param processInstance processInstance
     * @return task instance
     */
    public TaskInstance submitTaskInstanceToDB(TaskInstance taskInstance, ProcessInstance processInstance) {
        ExecutionStatus processInstanceState = processInstance.getState();

        if (taskInstance.getState().typeIsFailure()) {
            if (taskInstance.isSubProcess()) {
                taskInstance.setRetryTimes(taskInstance.getRetryTimes() + 1);
            } else {
                if (processInstanceState != ExecutionStatus.READY_STOP
                        && processInstanceState != ExecutionStatus.READY_PAUSE) {
                    // failure task set invalid
                    taskInstance.setFlag(Flag.NO);
                    updateTaskInstance(taskInstance);
                    // crate new task instance
                    if (taskInstance.getState() != ExecutionStatus.NEED_FAULT_TOLERANCE) {
                        taskInstance.setRetryTimes(taskInstance.getRetryTimes() + 1);
                    }
                    taskInstance.setSubmitTime(null);
                    taskInstance.setLogPath(null);
                    taskInstance.setExecutePath(null);
                    taskInstance.setStartTime(null);
                    taskInstance.setEndTime(null);
                    taskInstance.setFlag(Flag.YES);
                    taskInstance.setHost(null);
                    taskInstance.setId(0l);
                }
            }
        }
        taskInstance.setUserId(processInstance.getUserId());
        taskInstance.setProcessInstancePriority(processInstance.getProcessInstancePriority());
        taskInstance.setState(getSubmitTaskState(taskInstance, processInstanceState));
        if (taskInstance.getSubmitTime() == null) {
            taskInstance.setSubmitTime(new Date());
        }
        if (taskInstance.getFirstSubmitTime() == null) {
            taskInstance.setFirstSubmitTime(taskInstance.getSubmitTime());
        }
        boolean saveResult = saveTaskInstance(taskInstance);
        if (!saveResult) {
            return null;
        }
        return taskInstance;
    }

    /**
     * get submit task instance state by the work process state
     * cannot modify the task state when running/kill/submit success, or this
     * task instance is already exists in task queue .
     * return pause if work process state is ready pause
     * return stop if work process state is ready stop
     * if all of above are not satisfied, return submit success
     *
     * @param taskInstance taskInstance
     * @param processInstanceState processInstanceState
     * @return process instance state
     */
    public ExecutionStatus getSubmitTaskState(TaskInstance taskInstance, ExecutionStatus processInstanceState) {
        ExecutionStatus state = taskInstance.getState();
        // running, delayed or killed
        // the task already exists in task queue
        // return state
        if (
                state == ExecutionStatus.RUNNING_EXECUTION
                        || state == ExecutionStatus.DELAY_EXECUTION
                        || state == ExecutionStatus.KILL
        ) {
            return state;
        }
        //return pasue /stop if process instance state is ready pause / stop
        // or return submit success
        if (processInstanceState == ExecutionStatus.READY_PAUSE) {
            state = ExecutionStatus.PAUSE;
        } else if (processInstanceState == ExecutionStatus.READY_STOP
                || !checkProcessStrategy(taskInstance)) {
            state = ExecutionStatus.KILL;
        } else {
            state = ExecutionStatus.SUBMITTED_SUCCESS;
        }
        return state;
    }
    /**
     * check process instance strategy
     *
     * @param taskInstance taskInstance
     * @return check strategy result
     */
    private boolean checkProcessStrategy(TaskInstance taskInstance) {
        ProcessInstance processInstance = this.findProcessInstanceById(taskInstance.getProcessInstanceId());
        FailureStrategy failureStrategy = processInstance.getFailureStrategy();
        if (failureStrategy == FailureStrategy.CONTINUE) {
            return true;
        }
        List<TaskInstance> taskInstances = this.findValidTaskListByProcessId(taskInstance.getProcessInstanceId());

        for (TaskInstance task : taskInstances) {
            if (task.getState() == ExecutionStatus.FAILURE
                    && task.getRetryTimes() >= task.getMaxRetryTimes()) {
                return false;
            }
        }
        return true;
    }
    /**
     * format task app id in task instance
     */
    public String formatTaskAppId(TaskInstance taskInstance) {
        ProcessInstance processInstance = findProcessInstanceById(taskInstance.getProcessInstanceId());
        if (processInstance == null) {
            return "";
        }
        ProcessDefinition definition = findProcessDefinition(processInstance.getProcessDefinitionCode(), processInstance.getProcessDefinitionVersion());
        if (definition == null) {
            return "";
        }
        return String.format("%s_%s_%s", definition.getId(), processInstance.getId(), taskInstance.getId());
    }
    /**
     * find last scheduler process instance in the date interval
     *
     * @param definitionCode definitionCode
     * @param dateInterval dateInterval
     * @return process instance
     */
    public ProcessInstance findLastSchedulerProcessInterval(Long definitionCode, DateInterval dateInterval) {
        return processInstanceMapper.queryLastSchedulerProcess(definitionCode,
                dateInterval.getStartTime(),
                dateInterval.getEndTime());
    }

    /**
     * find last manual process instance interval
     *
     * @param definitionCode process definition code
     * @param dateInterval dateInterval
     * @return process instance
     */
    public ProcessInstance findLastManualProcessInterval(Long definitionCode, DateInterval dateInterval) {
        return processInstanceMapper.queryLastManualProcess(definitionCode,
                dateInterval.getStartTime(),
                dateInterval.getEndTime());
    }

    /**
     * find last running process instance
     *
     * @param definitionCode process definition code
     * @param startTime start time
     * @param endTime end time
     * @return process instance
     */
    public ProcessInstance findLastRunningProcess(Long definitionCode, Date startTime, Date endTime) {
        return processInstanceMapper.queryLastRunningProcess(definitionCode,
                startTime,
                endTime,
                stateArray);
    }

    /**
     * find sub process instance
     *
     * @param parentProcessId parentProcessId
     * @param parentTaskId parentTaskId
     * @return process instance
     */
    public ProcessInstance findSubProcessInstance(Long parentProcessId, Long parentTaskId) {
        ProcessInstance processInstance = null;
        ProcessInstanceMap processInstanceMap = processInstanceMapMapper.queryByParentId(parentProcessId, parentTaskId);
        if (processInstanceMap == null || processInstanceMap.getProcessInstanceId() == null) {
            return processInstance;
        }
        processInstance = findProcessInstanceById(processInstanceMap.getProcessInstanceId());
        return processInstance;
    }

    public DatasourceInfo findDataSourceById(long datasourceId) {
        return dataManageService.findDatasourceById(datasourceId);
    }

    public List<UdfDefined> queryUdfsByIds(List<Long> udfIds) {
        return taskService.findDatasourceById(udfIds);
    }

    public QueueInfo getQueue(Long queueId) {
        return queueInfoMapper.selectById(queueId);
    }

    public List<ResourceInfo> getResources(List<Long> resourceIds) {
        return taskService.findResByIds(resourceIds);
    }

    public Map<String,Set<PrivilegeType>> getAllTablePrivilege(Long datasourceId, Long proId, Long userId,Set<String> tables) {
        return dataManageService.getAllTablePrivilege(datasourceId,proId,userId,tables);
    }

    public String getUserKtUrl(String proName, String username) {
        return kbmService.getKeytabUrl(proName,username);
    }

    /**
     * check the input command exists in queue list
     *
     * @param command command
     * @return create command result
     */
    public boolean verifyIsNeedCreateCommand(Command command) {
        boolean isNeedCreate = true;
        EnumMap<CommandType, Integer> cmdTypeMap = new EnumMap<>(CommandType.class);
        cmdTypeMap.put(CommandType.REPEAT_RUNNING, 1);
        cmdTypeMap.put(CommandType.RECOVER_SUSPENDED_PROCESS, 1);
        cmdTypeMap.put(CommandType.START_FAILURE_TASK_PROCESS, 1);
        CommandType commandType = command.getCommandType();

        if (cmdTypeMap.containsKey(commandType)) {
            ObjectNode cmdParamObj = JSONUtils.parseObject(command.getCommandParam());
            Long processInstanceId = cmdParamObj.path(CMD_PARAM_RECOVER_PROCESS_ID_STRING).asLong();

            List<Command> commands = commandMapper.selectList(null);
            // for all commands
            for (Command tmpCommand : commands) {
                if (cmdTypeMap.containsKey(tmpCommand.getCommandType())) {
                    ObjectNode tempObj = JSONUtils.parseObject(tmpCommand.getCommandParam());
                    if (tempObj != null && processInstanceId.equals(tempObj.path(CMD_PARAM_RECOVER_PROCESS_ID_STRING).asLong())) {
                        isNeedCreate = false;
                        break;
                    }
                }
            }
        }
        return isNeedCreate;
    }

    /**
     * recursive query sub process definition id by parent id.
     *
     * @param parentCode parentCode
     * @param subProcessDefineCodes subProcessDefineCodes
     */
    public void recurseFindSubProcessCode(long parentCode, Set<Long> subProcessDefineCodes) {
        ProcessDefinition processDefinition = processDefineMapper.queryByCode(parentCode,0l);
        if (processDefinition == null) {
            logger.error("process define not exists");
            return;
        }
        Set<Long> subProcessCodes = new HashSet<>();
        List<TaskNodeParam> taskNodeParams = processDefinition.getTaskNodeParams();
        for (TaskNodeParam taskNodeParam : taskNodeParams) {
            if(taskNodeParam.getTaskType()== TaskType.SUB_PROCESS){
                Long processDefinitionCode = taskNodeParam.getSubProcessParameters().getProcessDefinitionCode();
                subProcessCodes.add(processDefinitionCode);
                subProcessDefineCodes.add(processDefinitionCode);
            }
        }
        for (Long subProcessCode : subProcessCodes) {
            recurseFindSubProcessCode(subProcessCode,subProcessDefineCodes);
        }
    }

    /**
     * query schedule by id
     *
     * @param id id
     * @return schedule
     */
    public Schedule querySchedule(long id) {
        return scheduleMapper.selectById(id);
    }

    public ProcessDefinition findProcessDefinitionByCode(long code) {
        return processDefineMapper.selectOne(new QueryWrapper<ProcessDefinition>().eq("code",code));
    }

    /**
     * Generate the DAG Graph based on the process definition id
     *
     * @param processDefinition process definition
     * @return dag graph
     */
    public DAG<String, TaskNode, TaskNodeRelation> genDagGraph(ProcessDefinition processDefinition) {
        List<ProcessTaskRelation> processTaskRelations = processTaskRelationMapper.selectList(new QueryWrapper<ProcessTaskRelation>().eq("process_definition_code", processDefinition.getCode()));
        List<TaskNode> taskNodeList = transformTask(processDefinition.getTaskNodeParams(),processTaskRelations, Maps.newHashMap());
        ProcessDag processDag = DagHelper.getProcessDag(taskNodeList, new ArrayList<>(processTaskRelations));
        // Generate concrete Dag to be executed
        return DagHelper.buildDagGraph(processDag);
    }

    public TableInfo getTableInfo(Long datasourceId, String dbName, String tableName) {
        return dataManageService.getTableInfo(datasourceId,dbName,tableName);
    }
}
